package com.rrd.dw.canal.client;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.List;

import org.apache.commons.lang.SystemUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.slf4j.MDC;
import org.springframework.util.Assert;
import org.springframework.util.CollectionUtils;

import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.protocol.CanalEntry.Column;
import com.alibaba.otter.canal.protocol.CanalEntry.Entry;
import com.alibaba.otter.canal.protocol.CanalEntry.EntryType;
import com.alibaba.otter.canal.protocol.CanalEntry.EventType;
import com.alibaba.otter.canal.protocol.CanalEntry.RowChange;
import com.alibaba.otter.canal.protocol.CanalEntry.RowData;
import com.alibaba.otter.canal.protocol.Message;
import com.rrd.dw.canal.sink.AbstractSink;

public class AbstractCanalClient {
	protected final static Logger logger = LoggerFactory.getLogger(AbstractCanalClient.class);
	protected static final String CRT001 = "\001";
	protected static final String SEP = SystemUtils.LINE_SEPARATOR;
	protected static final String DATE_FORMAT = "yyyy-MM-dd HH:mm:ss";
	protected volatile boolean running = false;
	protected Thread.UncaughtExceptionHandler handler = new Thread.UncaughtExceptionHandler() {

		public void uncaughtException(Thread t, Throwable e) {
			logger.error("parse events has an error", e);
		}
	};
	protected Thread thread = null;
	protected CanalConnector connector;
	protected static String context_format = null;
	protected static String row_format = null;
	protected static String transaction_format = null;
	protected String destination;
	protected AbstractSink sink = null;
	static {
		context_format = SEP + "****************************************************" + SEP;
		context_format += "* Batch Id: [{}] ,count : [{}] , memsize : [{}] , Time : {}" + SEP;
		context_format += "* Start : [{}] " + SEP;
		context_format += "* End : [{}] " + SEP;
		context_format += "****************************************************" + SEP;

		row_format = SEP
				+ "----------------> binlog[{}:{}] , name[{},{}] , eventType : {} , executeTime : {} , delay : {}ms"
				+ SEP;

		transaction_format = SEP + "================> binlog[{}:{}] , executeTime : {} , delay : {}ms" + SEP;

	}

	public AbstractCanalClient(String destination) {
		this(destination, null);
	}

	public AbstractCanalClient(String destination, CanalConnector connector) {
		this.destination = destination;
		this.connector = connector;
	}
	protected void start() {
		Assert.notNull(connector, "connector is null");
		thread = new Thread(new Runnable() {

			public void run() {
				process();
			}
		});

		thread.setUncaughtExceptionHandler(handler);
		thread.start();
		running = true;
	}

	protected void stop() {
		if (!running) {
			return;
		}
		running = false;
		if (thread != null) {
			try {
				thread.join();
			} catch (InterruptedException e) {
				// ignore
			}
		}

		MDC.remove("destination");
	}

	protected void process() {
		int batchSize = 5 * 1024;
		while (running) {
			try {
				MDC.put("destination", destination);
				
				while (running) {
					Message message = connector.getWithoutAck(batchSize); // 获取指定数量的数据
					long batchId = message.getId();
					int size = message.getEntries().size();
					if (batchId == -1 || size == 0) {
						try {
							Thread.sleep(1000);
						} catch (InterruptedException e) {
						}
					} else {
						
//						printSummary(message, batchId, size);
						parseMessage(message);
					}
					connector.ack(batchId); // 提交确认
					// connector.rollback(batchId); // 处理失败, 回滚数据
				}
			} catch (Exception e) {
				logger.error("process error!", e);
			} finally {
				connector.disconnect();
				MDC.remove("destination");
			}
		}
	}

	/**
	 * 解析message
	 * @param message
	 */
	protected void parseMessage(Message message) {
		List<Entry> entrys = message.getEntries();
		String opType=null;
		for (Entry entry : entrys) {
			if (entry.getEntryType() == EntryType.TRANSACTIONBEGIN
					|| entry.getEntryType() == EntryType.TRANSACTIONEND) {
				continue;
			} else if (entry.getEntryType() == EntryType.ROWDATA) {
				RowChange rowChage = null;
				try {
					rowChage = RowChange.parseFrom(entry.getStoreValue());
				} catch (Exception e) {
					throw new RuntimeException("parse event has an error , data:" + entry.toString(), e);
				}
				EventType eventType = rowChage.getEventType();
	                if (eventType == EventType.QUERY || rowChage.getIsDdl()) {
//	                    logger.info(" sql ----> " + rowChage.getSql() + SEP);
	                    System.out.println("#########################2eventType =="+eventType+"rowChage.getIsDdl()=="+rowChage.getIsDdl());
	                    
	                    continue;
	                }
	                for (RowData rowData : rowChage.getRowDatasList()) {
	                    if (eventType == EventType.DELETE) {
	                    	opType = "delete";
//	                    	rowDataToString(rowData.getBeforeColumnsList(),opType);
	                        continue;
	                    } else if (eventType == EventType.INSERT) {
	                    	opType="insert";
	                    	rowDataToString(rowData.getAfterColumnsList(),opType);
	                    } else {
	                    	opType="update";
//	                    	rowDataToString(rowData.getBeforeColumnsList(),opType);
	                    	rowDataToString(rowData.getAfterColumnsList(),opType);
	                    }
	                }
			}
		}
	}
	/**
	 * 解析rowdata，然后写入到存储
	 * @param rowData
	 * @param type
	 */
	private void rowDataToString(List<Column> columnList,String opType){
		StringBuilder sb = new StringBuilder(500);
		sb.append(opType);
		for(Column c : columnList){
			sb.append(CRT001).append(c.getValue()==null?"":c.getValue());
		}
		if(this.sink==null){
			printColumn(columnList);
		}else{
			this.sink.write(sb.toString().replace("\n", ""));
		}
		
	}
	protected void printSummary(Message message, long batchId, int size) {
		long memsize = 0;
		for (Entry entry : message.getEntries()) {
			memsize += entry.getHeader().getEventLength();
		}

		String startPosition = null;
		String endPosition = null;
		if (!CollectionUtils.isEmpty(message.getEntries())) {
			startPosition = buildPositionForDump(message.getEntries().get(0));
			endPosition = buildPositionForDump(message.getEntries().get(message.getEntries().size() - 1));
		}

		SimpleDateFormat format = new SimpleDateFormat(DATE_FORMAT);
		logger.info(context_format,
				new Object[] { batchId, size, memsize, format.format(new Date()), startPosition, endPosition });
	}

	protected String buildPositionForDump(Entry entry) {
		long time = entry.getHeader().getExecuteTime();
		Date date = new Date(time);
		SimpleDateFormat format = new SimpleDateFormat(DATE_FORMAT);
		return entry.getHeader().getLogfileName() + ":" + entry.getHeader().getLogfileOffset() + ":"
				+ entry.getHeader().getExecuteTime() + "(" + format.format(date) + ")";
	}
	
	protected static void printColumn(List<Column> columns) {
	        for (Column column : columns) {
	            System.out.println(column.getName() + " : " + column.getValue() + "    update=" + column.getUpdated());
	        }
	    }
	public void setConnector(CanalConnector connector) {
        this.connector = connector;
    }
	public void setSink(AbstractSink sink) {
        this.sink = sink;
    }
}
